package com.gitee.source.practice

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.kafka.clients.CommonClientConfigs

//TODO kafka的Source
object CustomerSourceKafka {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val props: Properties = kafkaConf

    val topic = "mytopic" //主题

    val data: DataStream[String] = kafkaConsumer(senv, props, topic)
    data.print()
    senv.execute()
  }

  def kafkaConsumer(senv: StreamExecutionEnvironment, props: Properties, topic: String): DataStream[String] = {
    //3.从Flink提供的自定义数据源FlinkKafkaConsumer011中获取数据
    //FlinkKafkaConsumer从kafka获取的每条消息都会通过DeserialzationSchema的T deserialize(byte[] message)反序列化处理
    //SimpleStringSchema可以将Kafka的字节消息反序列化为Flink的字符串对象
    //JSONDeserializationSchema(只反序列化value)/JSONKeyValueDeserializationSchema可以把序列化后的Json反序列化成ObjectNode，ObjectNode可以通过objectNode.get(“field”).as(Int/String/…)() 来访问指定的字段
    //TypeInformationSerializationSchema/TypeInformationKeyValueSerializationSchema基于Flink的TypeInformation来创建schema，这种反序列化对读写均是Flink的场景会比其他通用的序列化方式带来更高的性能。
    val kafkaConsumer = new FlinkKafkaConsumer011(topic, new SimpleStringSchema(), props)

    //4.还可以设置一下Kafka消费者的其他属性
    kafkaConsumer.setCommitOffsetsOnCheckpoints(true) //默认为true
    kafkaConsumer.setStartFromGroupOffsets() //默认值，从当前消费组记录的偏移量接着上次的开始消费,如果没有找到，则使用consumer的properties的auto.offset.reset设置的策略
    //kafkaConsumer.setStartFromEarliest()//从最早的数据开始消费
    //kafkaConsumer.setStartFromLatest//从最新的数据开始消费
    kafkaConsumer.setStartFromTimestamp(1568908800000L)//根据指定的时间戳消费数据
    /*val offsets = new util.HashMap[KafkaTopicPartition, java.lang.Long]()//key是KafkaTopicPartition(topic，分区id),value是偏移量
    offsets.put(new KafkaTopicPartition(topic, 0), 110L)
    offsets.put(new KafkaTopicPartition(topic, 1), 119L)
    offsets.put(new KafkaTopicPartition(topic, 2), 120L)*/
    //kafkaConsumerSource.setStartFromSpecificOffsets(offsets)// TODO 从指定的具体位置开始消费,这个设置可以将偏移量提交到mysql
    import org.apache.flink.api.scala._
    val data: DataStream[String] = senv.addSource(kafkaConsumer)
    data
  }

  def kafkaConf = {
    // 准备Kafka连接参数
    val props = new Properties()
    props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "node01:9092")
    props.setProperty("group.id", "mytopic")
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    /*
    earliest 有提交,从已提交的offset开始消费,无提交从头开始消费
    latest 有提交,从已提交的offset开始消费,无提交从消费新产生该分区下的数据开始消费
    none 各个分区都存在已提交的offset从已提交的offset开始消费,只要有一个分区不存在已提交的offset就抛出异常
    注意: latest这个设置容易丢失数据, 假如kafka出现问题,这个时候还有数据往topic中写,那么重启kafka,这个设置会从最新的offset开始消费,中间出故障的一段数据就不管了
     */
    props.setProperty("auto.offset.reset", "latest")
    //动态感知kafka主题分区的变化,开启一个后台线程每隔5s查询一次最新的主题分区变化
    props.setProperty("flink.partition-discovery.interval-millis", "5000")
    props
  }
}
